// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-11-19

package common

import (
	"context"
	"gitee.com/go-libs/runtime"
	"sync"
	"sync/atomic"
	"time"
)

const (
	bucketMaxCapacity = 30000
)

type (
	Bucket[T any] interface {
		Add(list ...T)
		Count() (count int)
		GetBatch() (batch int)
		GetConcurrency() (concurrency int32)
		GetFrequency() (frequency int)
		Name() string
		SetBatch(batch int)
		SetConcurrency(concurrency int32)
		SetFrequency(frequency int)
		Start()
		Stop() Bucket[T]
		Wait()
	}

	bucket[T any] struct {
		handler func(...T)
		list    []T
		mu      *sync.Mutex
		name    string

		cancel  context.CancelFunc
		ctx     context.Context
		started bool

		batch, frequency        int
		concurrency, processing int32
		total                   int
	}
)

// NewBucket
// 创建数据桶.
func NewBucket[T any](name string, handler func(...T)) Bucket[T] {
	return &bucket[T]{
		handler: handler,
		list:    make([]T, 0),
		mu:      &sync.Mutex{},
		name:    name,

		batch: 300, frequency: 350,
		concurrency: 3,
	}
}

// Add
// 添加元素.
func (o *bucket[T]) Add(list ...T) {
	if n := len(list); n > 0 {
		o.mu.Lock()

		// 丢弃日志避免内存溢出.
		if (o.total + n) >= bucketMaxCapacity {
			o.mu.Unlock()
			runtime.GetCounter().LoggingCounter().IncrDiscardCount(int64(n))
			return
		}

		o.list = append(o.list, list...)
		o.total = len(o.list)
		o.mu.Unlock()
		runtime.GetCounter().LoggingCounter().IncrPushCount(int64(n))

		// 立即取出, 若内存中积压的数据量大于或等于批次数, 则立即处理.
		if o.total >= o.batch {
			go o.pop()
		}
	}
}

func (o *bucket[T]) Count() (count int) {
	o.mu.Lock()
	count = len(o.list)
	o.mu.Unlock()
	return
}

func (o *bucket[T]) GetBatch() (batch int) {
	o.mu.Lock()
	batch = o.batch
	o.mu.Unlock()
	return
}

func (o *bucket[T]) GetConcurrency() (concurrency int32) {
	o.mu.Lock()
	concurrency = o.concurrency
	o.mu.Unlock()
	return
}

func (o *bucket[T]) GetFrequency() (frequency int) {
	o.mu.Lock()
	frequency = o.frequency
	o.mu.Unlock()
	return
}

func (o *bucket[T]) Name() string { return o.name }

func (o *bucket[T]) SetBatch(batch int) {
	if batch > 0 {
		o.mu.Lock()
		o.batch = batch
		o.mu.Unlock()
	}
}

func (o *bucket[T]) SetConcurrency(concurrency int32) {
	if concurrency > 0 {
		o.mu.Lock()
		o.concurrency = concurrency
		o.mu.Unlock()
	}
}

func (o *bucket[T]) SetFrequency(frequency int) {
	if frequency > 0 {
		o.mu.Lock()
		o.frequency = frequency
		o.mu.Unlock()
	}
}

func (o *bucket[T]) Start() {
	o.mu.Lock()

	if o.started {
		o.mu.Unlock()
		return
	}

	o.ctx, o.cancel = context.WithCancel(context.Background())
	o.started = true
	o.mu.Unlock()
	go o.start()
}

func (o *bucket[T]) Stop() Bucket[T] {
	o.mu.Lock()
	defer o.mu.Unlock()

	if o.ctx != nil && o.ctx.Err() == nil {
		o.cancel()
	}
	return o
}

func (o *bucket[T]) Wait() {
	d, s, t := time.Second*5, time.Millisecond*10, time.Now()
	for {
		if atomic.LoadInt32(&o.processing) <= 0 && func() bool {
			o.mu.Lock()
			defer o.mu.Unlock()
			return len(o.list) < 1
		}() {
			break
		}

		if time.Now().Sub(t) >= d {
			break
		}
		time.Sleep(s)
	}
}

// +------------------------------------------------------------------------------+
// | Access methods                                                               |
// +------------------------------------------------------------------------------+

func (o *bucket[T]) pop() {
	var (
		list  []T
		count int
	)

	// 1. 并发限流.
	if n := atomic.AddInt32(&o.processing, 1); n > o.concurrency {
		atomic.AddInt32(&o.processing, -1)
		return
	}

	// 2. 监听结束.
	defer func() {
		// 2.1 捕获异常.
		if r := recover(); r != nil {
			Debugger.Fatal("bucket pop panic: %v", r)
		}

		// 2.2 恢复记数.
		atomic.AddInt32(&o.processing, -1)

		// 2.3 继续处理.
		if o.Count() > 0 {
			o.pop()
		}
	}()

	// 3. 取出数据.
	o.mu.Lock()

	// 3.1 列表为空.
	if o.total = len(o.list); o.total < 1 {
		o.mu.Unlock()
		return
	}

	// 3.2 取出数据.
	if o.total > o.batch {
		// 片段取出.
		count = o.batch
		list = o.list[0:count]
		o.list = o.list[count:]
		o.total -= count
	} else {
		// 取出全部.
		count = o.total
		list = o.list[0:count]
		o.list = make([]T, 0)
		o.total = 0
	}
	o.mu.Unlock()
	runtime.GetCounter().LoggingCounter().IncrPopCount(int64(count))

	// 3.3 处理数据.
	o.handler(list...)
}

func (o *bucket[T]) start() {
	var (
		clear = func() {
			for i := 0; i < int(atomic.LoadInt32(&o.concurrency)); i++ {
				go o.pop()
			}
		}
		ticker = time.NewTicker(time.Millisecond * time.Duration(o.frequency))
	)

	// 1. 清理数据.
	//    数据桶退出前, 清理内存中积压的数据
	defer clear()

	// 2. 监听信号.
	for {
		select {
		case <-ticker.C:
			go o.pop()
		case <-o.ctx.Done():
			return
		}
	}
}
